package com.dream.cms.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dream.cms.config.KafkaTopicProperties;
import com.dream.cms.dao.LogRepository;
import com.dream.cms.dao.LogRepositoryDymi;
import com.dream.cms.entity.Log;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class LogKafkaConsumer {


    @Autowired
    private KafkaTopicProperties properties;

    @Autowired
    private LogRepository logRepository;

    @Autowired
    private LogRepositoryDymi logRepositoryDymi;

    public void add(Log log){
        logRepositoryDymi.insertLog(null);
    }
//,concurrency = "6",containerFactory="kafkaListenerContainerFactory"
    @KafkaListener(topics = "${custom.topic-name}",containerFactory="batchFactory")
    private void task(List<ConsumerRecord<String, String>> records){
        batchConsumer(records);
        }

    /**
     * 单条消费
     */
    public void consumer(ConsumerRecord<String, String> record) {
        Log logEntity = this.convertToLog(record.value());
        if(logEntity != null){
            logRepository.save(logEntity);
        }
    }
    public Log convertToLog(String log111) {
        Log logEntity = null;
        try {
//            JSONObject item = JSONObject.parseObject(log111);
//            JSONObject message = JSON.parseObject(item.getString("message"));
//            String request = message.getString("request");
//            if (this.needToLog(request)) {
                logEntity = new Log();
                logEntity.setContent(log111);
//                JSONObject beat = item.getJSONObject("beat");
//                logEntity.setHostName(beat.getString("hostname"));
//                logEntity.setNginxIp(item.getString("nginx_ip"));
//                logEntity.setBodyBytesSent(message.getString("body_bytes_sent"));
//                logEntity.setClientRealIp(message.getString("clientRealIp"));
//                String time = this.convertDate(message.getString("@timestamp"));
//                logEntity.setTime(time);
//                logEntity.setHttpUserAgent(message.getString("http_user_agent"));
//                logEntity.setStatus(message.getString("status"));
//                logEntity.setBodyBytesSent(message.getString("body_bytes_sent"));
//                logEntity.setHttpReferer(message.getString("http_referer"));
//                logEntity.setRemoteAddr(message.getString("remote_addr"));
//                logEntity.setRemoteUser(message.getString("remote_user"));
//                logEntity.setRequest(message.getString("request"));
//                logEntity.setUpstreamAddr(message.getString("upstream_addr"));
//                logEntity.setUpstream_status(message.getString("upstream_status"));
//                logEntity.setUpstreamResponseLength(message.getString("upstream_response_length"));
//                logEntity.setRequestTime(message.getString("request_time"));
//                logEntity.setUpstreamResponseTime(message.getString("upstream_response_time"));
//            }
        }catch(Exception e){
            log.error(e.getMessage());
            log.error(log111);
        }
        return logEntity;
    }

    /**
     * 批量消费
     */
    public void batchConsumer(List<ConsumerRecord<String, String>> records) {
        log.debug("wgq records size is:"+ records.size());
        List<Log> logs = new ArrayList<>();
        records.forEach(record -> {
            Log logEntity = this.convertToLog(record.value());
            if(logEntity != null){
                logs.add(logEntity);
            }
        });
        if(!CollectionUtils.isEmpty(logs)){
            logRepositoryDymi.insertLog(logs);
        }
    }
    private boolean needToLog(String url){
        boolean flag = false;
        String[] pages = properties.getPages();
        if(pages != null && pages.length > 0){
            for(int i =0,len = pages.length;i<len;i++){
                if(url.contains(pages[i])){
                    flag = true;
                    break;
                }
            }
        }
        return flag;
    }
    private String convertDate(String date){
        DateTime dt = new DateTime( date ) ;
        String result = dt.toString("yyyy-MM-dd hh:mm:ss");
        return result;
    }
}
